package org.unigram.oluolu.recommend;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.unigram.oluolu.common.OluoluConstants;
import org.unigram.oluolu.common.OluoluException;
import org.unigram.oluolu.common.Query;
import org.unigram.oluolu.extractor.IExtractor;

/**
 * Change input format into feature representation.
 */
public class ChangeFormatMapper extends
    Mapper<LongWritable, Text, Text, Query> {

    /** extractor to get Query from each line. */
    private IExtractor extractor;    
    
    
    /**
     * map function.
     * 
     * @param key dummy
     * @param value contains user id, time, query string  
     * @param context write key value 
     * 
     * @throws IOException -
     * @throws InterruptedException -
     */
    @Override
    public final void map(final LongWritable key,
            final Text value, final Context context)
            throws IOException, InterruptedException {
        try {
            Query query = extractor.extract(value.toString());
            context.write(query.getUserId(), query);
        } catch (OluoluException e) {
            e.printStackTrace();
        }
    }
    
    /**
     * Build Extractor object which is used to create Query objects 
     * from input string. 
     * 
     * @param context containing configuration object which 
     *         has the name of extractor used in this object.
     */
    public final void setup(final Context context) {
        Configuration jc = context.getConfiguration();
        String extractorClassName = OluoluConstants.DEFAULT_EXTRACTOR;
        if (context == null || jc == null) {
            /* added default configuration for testing */
            jc = new Configuration();
        }
        try {
            extractorClassName = jc.get(OluoluConstants.EXTRACTOR,
                    OluoluConstants.DEFAULT_EXTRACTOR);
            Class<? extends IExtractor> extractorClass = Class.forName(
                    extractorClassName).asSubclass(IExtractor.class);
            Constructor<? extends IExtractor> constructor = extractorClass
                    .getConstructor(Configuration.class);
            extractor = constructor.newInstance(jc);
        } catch (NoSuchMethodException nsme) {
            throw new RuntimeException(nsme);
        } catch (ClassNotFoundException cnfe) {
            throw new RuntimeException(cnfe);
        } catch (InstantiationException ie) {
            throw new RuntimeException(ie);
        } catch (IllegalAccessException iae) {
            throw new RuntimeException(iae);
        } catch (InvocationTargetException ite) {
            throw new RuntimeException(ite.getCause());
        }
    }    
}
